package org.luckyjourney.service.impl;

import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.luckyjourney.constant.RedisConstant;
import org.luckyjourney.entity.video.Video;
import org.luckyjourney.entity.user.User;
import org.luckyjourney.entity.vo.HotVideo;
import org.luckyjourney.entity.vo.Model;
import org.luckyjourney.entity.vo.UserModel;
import org.luckyjourney.exception.BaseException;
import org.luckyjourney.service.InterestPushService;
import org.luckyjourney.service.user.UserService;
import org.luckyjourney.service.video.TypeService;
import org.luckyjourney.util.BloomFilterUtil;
import org.luckyjourney.util.RedisCacheUtil;
import org.luckyjourney.util.TypeIdGenerator;
import org.redisson.api.RBloomFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @description:
 * @Author: Xhy
 * @CreateTime: 2023-10-26 11:54
 */
// 暂时为异步
@Service
public class InterestPushServiceImpl implements InterestPushService {

    @Autowired
    private RedisCacheUtil redisCacheUtil;

    @Autowired
    private TypeService typeService;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private TypeIdGenerator typeIdGenerator;

    final ObjectMapper objectMapper = new ObjectMapper();

    @Autowired
    private BloomFilterUtil bloomFilterUtil;


    @Override
    @Async
    public void pushSystemStockIn(Video video) {
        // 往系统库中添加
        final List<String> labels = video.buildLabel();
        final Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sAdd((RedisConstant.SYSTEM_STOCK + label).getBytes(),String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    @Override
    @Async
    // 实现分片存储
    public void pushSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        String key = getTypeStockKey(typeId, typeIdGenerator.getAndIncre(typeId));
        redisCacheUtil.sSet(key,video.getId());
    }

    @Override
    public Collection<Long> listVideoIdByTypeId(Long typeId) {
        // 从3个分片中随机推送10个视频
        final HashSet<Long> result = new HashSet<>();
        Long id = typeIdGenerator.getId(typeId);
        if(id == null){
            return null;
        }
        List<Object> list;
        int videoNum = 8;
        // 若当前分片id存储数较小，则取上一分片的视频id
        // 防止死循环
        while (result.size() < 12 && id >= 0){
            // 推送大部分新视频，小部分老视频
            String key = getTypeStockKey(typeId, id);
            Random random = new Random(System.currentTimeMillis());
            if(id > 0) {
                id = (long) random.nextInt(id.intValue());
            }
            else id--;
            // 在获取视频之前迭代，保证id为0时只执行一次循环，同时在视频数量较少时不造成死循环
            list = redisTemplate.opsForSet().randomMembers(key, videoNum);
            if (!ObjectUtils.isEmpty(list)) {
                for (Object aLong : list) {
                    if (aLong != null){
                        result.add(Long.parseLong(aLong.toString()));
                    }
                }
            }
            if(videoNum <= result.size()){
                // 正常迭代
                videoNum = videoNum / 3;
            }
        }
        return result;
    }

    private static String getTypeStockKey(Long typeId, Long id) {
        String key = RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + id;
        return key;
    }

    @Override
    @Async
    public void deleteSystemStockIn(Video video) {
        final List<String> labels = video.buildLabel();
        final Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sRem((RedisConstant.SYSTEM_STOCK + label).getBytes(),String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    @Override
    @Async
    public void initUserModel(Long userId, List<String> labels) {
        /*
            思考：为什么使用redis？
                首先，用户兴趣模型是频繁修改更新的，
                其次，兴趣模型结构为hashmap，key为标签，value为概率，不适合mysql存储结构
                而且业务需求中有随机数的需求，mysql无法实现
        */
        final String key = RedisConstant.USER_MODEL + userId;
        // key:标签名 value:标签被推送概率的百分比数值（均分）
        Map<String, Object> modelMap = new HashMap<>();
        if (!ObjectUtils.isEmpty(labels)) {
            final int size = labels.size();
            // 将标签分为等分概率,不可能超过100个分类
            double probabilityValue = 100 / size;
            for (String labelName : labels) {
                modelMap.put(labelName, probabilityValue);
            }
        }
        // redisCacheUtil.del(key);
        // 为用户模型设置ttl
        redisCacheUtil.hmset(key, modelMap, RedisConstant.USER_MODAL_TIME);
    }


    /**
     * 更新用户兴趣模型
     * @param  * 参数 userModel
     * @return
    */
    @Override
    @Async
    public void updateUserModel(UserModel userModel) {
        final Long userId = userModel.getUserId();
        // 游客不用管
        if (userId != null) {
            final List<Model> models = userModel.getModels();
            // 获取用户模型
            String key = RedisConstant.USER_MODEL + userId;
            Map<Object, Object> modelMap = redisCacheUtil.hmget(key);

            if (modelMap == null) {
                modelMap = new HashMap<>();
            }
            for (Model model : models) {
                // 修改用户模型
                if (modelMap.containsKey(model.getLabels())) {
                    modelMap.put(model.getLabels(), Double.parseDouble(modelMap.get(model.getLabels()).toString()) + model.getScore());
                    final Object o = modelMap.get(model.getLabels());
                    if (o == null || Double.parseDouble(o.toString()) <= 0.0){
                        modelMap.remove(o);
                    }
                } else {
                    modelMap.put(model.getLabels(), model.getScore());
                }
            }
            // 每个标签概率同等加上标签数，再同等除以标签数  防止数据膨胀
            final int labelSize = modelMap.keySet().size();
            for (Object o : modelMap.keySet()) {
                modelMap.put(o,(Double.parseDouble(modelMap.get(o).toString()) + labelSize )/ labelSize);
            }
            // 更新用户模型并设置TTL
            redisCacheUtil.hmset(key, modelMap, RedisConstant.USER_MODAL_TIME);
        }

    }

    /**
     * 兴趣推送视频给用户
     * @param  * 参数 user
     * @return
    */
    @Override
    public Collection<Long> listVideoIdByUserModel(User user) {
        /*
            思考：一次推送10条的原因
                推送太少，服务器压力大
                推送太多，不利于及时根据用户兴趣模型更新推送的视频类型，用户体验较差
        */
        // 创建结果集 默认推送10条视频
        Set<Long> videoIds = new HashSet<>(10);
        // 登录状态
        if (user != null) {
            final Long userId = user.getId();
            // 取出用户兴趣模型
            final Map<Object, Object> modelMap = redisCacheUtil.hmget(RedisConstant.USER_MODEL + userId);
            if (!ObjectUtils.isEmpty(modelMap)) {
                // 组成数组
                final String[] probabilityArray = initProbabilityArray(modelMap);
                final Boolean sex = user.getSex();
                final Random randomObject = new Random();
                final ArrayList<String> labelNames = new ArrayList<>();
                // 随机获取X个视频
                for (int i = 0; i < 8; i++) {
                    String labelName = probabilityArray[randomObject.nextInt(probabilityArray.length)];
                    labelNames.add(labelName);
                }
                // 标签对应的视频存在redis中， key: SYSTEM_STOCK:标签名  value: video_id
                String key1 = RedisConstant.SYSTEM_STOCK;
                // 随机获取 管道操作允许多个 Redis 命令一次性发送给 Redis 服务器，减少网络延迟，提高性能。
                List<Object> list = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (String labelName : labelNames) {
                        String key = key1 + labelName;
                        connection.sRandMember(key.getBytes());
                    }
                    return null;
                });
                // 获取到的videoIds
                Set<Long> vIds = list.stream().filter(id->id!=null).map(id->Long.parseLong(id.toString())).collect(Collectors.toSet());
                String key2 = RedisConstant.HISTORY_VIDEO;
                // 去重：不再推送已经观看过的视频
                // 采用布隆过滤器来实现浏览记录去重
                vIds = vIds.stream().filter(id -> !bloomFilterUtil.contains(userId, id)).collect(Collectors.toSet());
                videoIds.addAll(vIds);
                /*
                    思考：除了推送用户订阅的分类包括的默认标签，还可以推送什么标签？
                            推送相关标签（用户可能感兴趣的标签）
                                例： 动漫 -> 二游
                */
                // 随机挑选一个视频,根据性别: 男：美女 女：宠物
                final Long aLong = randomVideoId(sex);
                if (aLong!=null){
                    videoIds.add(aLong);
                }
                return videoIds;
            }
        }
        // 游客 随机获取10个标签进行推送
        final List<String> labels = typeService.random10Labels();
        final ArrayList<String> labelKeys = new ArrayList<>();
        int size = labels.size();
        final Random random = new Random();
        // 拼接标签视频库的key
        for (int i = 0; i < 10; i++) {
            final int randomIndex = random.nextInt(size);
            labelKeys.add(RedisConstant.SYSTEM_STOCK + labels.get(randomIndex));
        }
        // 从Redis当中获取videoId
        final List<Object> list = redisCacheUtil.sRandom(labelKeys);
        if (!ObjectUtils.isEmpty(list)){
            videoIds = list.stream().filter(id ->!ObjectUtils.isEmpty(id)).map(id -> Long.valueOf(id.toString())).collect(Collectors.toSet());
        }
        return videoIds;
    }

    @Override
    public Collection<Long> listVideoIdByLabels(List<String> labelNames) {
        final ArrayList<String> labelKeys = new ArrayList<>();
        for (String labelName : labelNames) {
            labelKeys.add(RedisConstant.SYSTEM_STOCK + labelName);
        }
        Set<Long> videoIds = new HashSet<>();
        final List<Object> list = redisCacheUtil.sRandom(labelKeys);
        if (!ObjectUtils.isEmpty(list)){
            videoIds = list.stream().filter(id ->!ObjectUtils.isEmpty(id)).map(id -> Long.valueOf(id.toString())).collect(Collectors.toSet());
        }
        return videoIds;
    }

    @Override
    @Async
    public void deleteSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        // redisCacheUtil.setRemove(RedisConstant.SYSTEM_TYPE_STOCK + typeId,video.getId());
        // 遍历所有分片
        Long id = typeIdGenerator.getId(typeId);
        do{
            String key = getTypeStockKey(typeId, id--);
            Boolean b = redisTemplate.opsForSet().isMember(key, video.getId());
            if(b != null && b == true){
                redisTemplate.opsForSet().remove(key, video.getId());
                break;
            }
        }while (id >= 0);
    }



    public Long randomHotVideoId(){
        final Object o = redisTemplate.opsForZSet().randomMember(RedisConstant.HOT_RANK);
        try {
            return objectMapper.readValue(o.toString(),HotVideo.class).getVideoId();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return null;
    }

    public Long randomVideoId(Boolean sex) {
        String key = RedisConstant.SYSTEM_STOCK + (sex ? "美女" : "宠物");
        final Object o = redisCacheUtil.sRandom(key);
        if (o!=null){
            return Long.parseLong(o.toString());
        }
        return null;
    }

    // 随机获取视频id
    public Long getVideoId(Random random, String[] probabilityArray) {
        String labelName = probabilityArray[random.nextInt(probabilityArray.length)];
        // 获取对应所有视频
        String key = RedisConstant.SYSTEM_STOCK + labelName;
        final Object o = redisCacheUtil.sRandom(key);
        if (o!=null){
            return Long.parseLong(o.toString()) ;
        }
        return null;
    }


    /**
     * 生成一个标签名出现次数和推送概率成正比的数组
     * @param  * 参数 modelMap 原始的用户兴趣模型
     * @return
    */
    public String[] initProbabilityArray(Map<Object, Object> modelMap) {
        // key: 标签  value：处理后的概率
        Map<String, Integer> probabilityMap = new HashMap<>();
        int size = modelMap.size(); // 标签数量
        // n：累加器，记录概率总和
        final AtomicInteger n = new AtomicInteger(0);
        modelMap.forEach((k, v) -> {
            // 防止结果为0,每个同等加上标签数  同比概率增减
            // v:标签名对应的概率百分比值（0~100）
            int probability = (((Double) v).intValue() + size) / size;
            n.getAndAdd(probability);
            probabilityMap.put(k.toString(), probability);
        });
        final String[] probabilityArray = new String[n.get()];
        // 每个标签开始的数组下标
        final AtomicInteger index = new AtomicInteger(0);
        // 初始化数组
        probabilityMap.forEach((labelName, p) -> {
            int i = index.get();
            // p为数组中一个标签所占的区间长度
            int limit = i + p;
            while (i < limit) {
                probabilityArray[i++] = labelName;
            }
            // 为新标签更新区间起始点
            index.set(limit);
        });
        return probabilityArray;
    }

}
